package com.bw.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName: DataGeneratorDemo
 * @Author: huzx
 * @Date: 2025-01-03 22:58
 * @Version: 1.0
 **/
public class DataGeneratorDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataGeneratorSource<String> ds = new DataGeneratorSource<>(
                // 指定 GeneratorFunction 实现类
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long value) throws Exception {
                        return "number:"+value;
                    }
                },
                // 指定 输出数据的总行数，自动生成的数字序列（从0自增）的最大值(小于)，达到这个值就停止了
                Long.MAX_VALUE,
                // 指定 每秒发射的记录数，限速策略， 比如 每秒生成几条数据
                RateLimiterStrategy.perSecond(10),
                // 指定返回值类型
                Types.STRING
        );
        env.fromSource(
                ds,
                WatermarkStrategy.noWatermarks(),
                "ds"
        ).print();

        env.execute();
    }
}


